package top.jolyoulu.core.example

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 * @Author: JolyouLu
 * @Date: 2024/2/6 16:32
 * @Description
 * 编号	字段名称	字段类型	字段含义
 * 1	date	String	用户点击行为的日期
 * 2	user_id	Long	用户的 ID
 * 3	session_id	String	Session 的 ID
 * 4	page_id	Long	某个页面的 ID
 * 5	action_time	String	动作的时间点
 * 6	search_keyword	String	用户搜索的关键词
 * 7	click_category_id	Long	某一个商品品类的 ID
 * 8	click_product_id	Long	某一个商品的 ID
 * 9	order_category_ids	String	一次订单中所有品类的 ID 集合
 * 10	order_product_ids	String	一次订单中所有商品的 ID 集合
 * 11	pay_category_ids	String	一次支付中所有品类的 ID 集合
 * 12	pay_product_ids	String	一次支付中所有商品的 ID 集合
 * 13	city_id	Long	城市 id
 */
object Spark01_Example_HotCategoryTop10SessionAnalysis {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
    val sc: SparkContext = new SparkContext(sparkConf)

    //读取日志数据
    val path: String = this.getClass.getClassLoader.getResource("datas/user_visit_action.txt").toURI.getPath
    val actionRDD: RDD[String] = sc.textFile(path)
    actionRDD.cache()
    //获取前10品类
    val top10Ids: Array[String] = top10Category(actionRDD)
    //过滤原始数据，保留前10品类id
    val filterActionRDD: RDD[String] = actionRDD.filter(
      action => {
        val datas = action.split("_")
        if (datas(6) != "-1") {
          top10Ids.contains(datas(6))
        } else {
          false
        }
      }
    )

    //根据品类id和sessionid根据点击量统计
    val reduceRDD: RDD[((String, String), Int)] = filterActionRDD.map(
      action => {
        val datas = action.split("_")
        ((datas(6), datas(2)), 1)
      }
    ).reduceByKey(_ + _)

    //将统计结果进行结构转换
    //((品类ID,sessionId),sum) => (品类ID,(sessionId,sum))
    val mapRDD: RDD[(String, (String, Int))] = reduceRDD.map {
      case ((cid, sid), sum) => {
        (cid, (sid, sum))
      }
    }

    //相同品类分组
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = mapRDD.groupByKey()

    //将分组后的数据进行点击量排序取前10
    val resultRDD: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
      }
    )

    resultRDD.collect().foreach(println)

    sc.stop()
  }


  def top10Category(actionRDD: RDD[String]): Array[String] = {
    //将数据结构转换，3种
    // (品类ID,(点击数量,下单数量,支付数量))
    // (品类ID,(1,0,0))
    // (品类ID,(0,1,0))
    // (品类ID,(0,0,1))
    val flatRDD: RDD[(String, (Int, Int, Int))] = actionRDD.flatMap(
      action => {
        val datas: Array[String] = action.split("_")
        //点击
        if (datas(6) != "-1") {
          List((datas(6), (1, 0, 0)))
          //下单
        } else if (datas(8) != "null") {
          val ids: Array[String] = datas(8).split(",")
          ids.map(id => (id, (0, 1, 0)))
          //支付
        } else if (datas(10) != "null") {
          val ids: Array[String] = datas(10).split(",")
          ids.map(id => (id, (0, 0, 1)))
        } else {
          Nil
        }
      }
    )

    //聚合
    val analysisRDD: RDD[(String, (Int, Int, Int))] = flatRDD.reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    )

    analysisRDD.sortBy(_._2, false).take(10).map(_._1)
  }
}